#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <unistd.h>
#include <math.h>
#include "MQTTClient.h"
#include "cJSON.h"
#include "shmem.h"
#include "list.h"
#include <modbus.h>
#include <sys/ipc.h>
#include "msg_queue_peer.h"
#include <errno.h>
#include <time.h>
#include <sqlite3.h>

#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "GatewayReport"
#define TOPIC_UP "up"
#define TOPIC_DOWN "down"
#define QOS 1
#define TIMEOUT 10000L
#define CONFIG_PATH "/mnt/config/dianbiao.json"
#define DB_PATH "/mnt/data/history.db"
#define MAX_NODES 100

// 上报模式
typedef enum
{
    REPORT_NONE = 0,   // 不上报
    REPORT_CHANGE = 1, // 变化上报
    REPORT_TIMED = 2   // 周期上报
} ReportMode;

// 设备类型
typedef enum
{
    DEV_STM32 = 1,
    DEV_MODBUS = 2,
} DeviceType;

// 共享内存数据结构
typedef union val_t
{
    int b_val;   // bool类型存储空间
    int i_val;   // 整形值存储空间
    float f_val; // 浮点值存储空间
} val_t;

struct std_node
{
    int key;       // 唯一键值
    int type;      // 数据点类型：1-bool, 2-int, 3-float
    int dev_type;  // 设备类型（例如：0-STM32, 1-Modbu）
    val_t old_val; // 旧值（用于变化检测）
    val_t new_val; // 新值（从设备采集的最新值）
    int ret;       // 采集状态：-1-失败，0-成功
};

// 全局配置结构
struct GlobalConfig
{
    int version;                // 协议版本
    ReportMode report_type;     // 上报类型
    int report_interval;        // 上报间隔(秒)
    char mqtt_addr[64];         // MQTT服务器地址
    int mqtt_port;              // MQTT端口
    char mb_dev_addr[64];       // modbus设备地址
    int mb_dev_port;            // modbus设备端口
    int node_count;             // 数据点数量
    struct std_node *shm_nodes; // 指向共享内存中的节点数组
    struct shm_param shm_para;  // 共享内存参数
    MQTTClient mqtt_client;     // MQTT客户端
    pthread_mutex_t lock;       // 线程锁
};

// 指令类型
typedef enum
{
    CMD_CONTROL,     // 控制命令
    CMD_REFRESH,     // 刷新上报
    CMD_CHANGE_MODE, // 修改上报模式
    CMD_HISTORY_DATA // 请求历史数据
} CommandType;

// 函数声明
void *reporting_thread(void *arg);
void *storage_thread(void *arg);
void *command_thread(void *arg);
int parse_config(struct GlobalConfig *config, struct std_node **nodes, int *node_count);
void send_mqtt_message(struct GlobalConfig *config, cJSON *data);
void handle_command(struct GlobalConfig *config, const char *command);
int init_database();
int store_data_point(sqlite3 *db, struct std_node *node);
cJSON *get_history_data(time_t start, time_t end, int key);

// 全局配置实例
struct GlobalConfig *config = NULL;
struct shm_param shm_para;

// 上报线程实现
void *reporting_thread(void *arg)
{
    struct GlobalConfig *cfg = (struct GlobalConfig *)arg;
    int node_count = cfg->node_count;
    struct std_node *nodes = cfg->shm_nodes;

    // 如果是变化上报，先获取初始值作为基准
    if (cfg->report_type == REPORT_CHANGE)
    {
        for (int i = 0; i < node_count; i++)
        {
            nodes[i].old_val = nodes[i].new_val;
        }
    }

    while (1)
    {
        // 如果是不上报模式，则跳过
        if (cfg->report_type == REPORT_NONE)
        {
            sleep(1);
            continue;
        }

        pthread_mutex_lock(&cfg->lock);

        cJSON *report = cJSON_CreateArray();

        // 根据不同上报模式处理
        if (cfg->report_type == REPORT_TIMED)
        {
            // 定时上报 - 上报所有数据点
            for (int i = 0; i < node_count; i++)
            {
                cJSON *item = cJSON_CreateObject();
                cJSON_AddNumberToObject(item, "key", nodes[i].key);

                if (nodes[i].type == 1)
                {
                    cJSON_AddBoolToObject(item, "value", nodes[i].new_val.b_val);
                }
                else if (nodes[i].type == 2)
                {
                    cJSON_AddNumberToObject(item, "value", nodes[i].new_val.i_val);
                }
                else
                {
                    cJSON_AddNumberToObject(item, "value", nodes[i].new_val.f_val);
                }

                cJSON_AddItemToArray(report, item);
            }
        }
        else if (cfg->report_type == REPORT_CHANGE)
        {
            // 变化上报 - 只上报变化的数据点
            for (int i = 0; i < node_count; i++)
            {
                int changed = 0;

                if (nodes[i].type == 1)
                {
                    changed = (nodes[i].new_val.b_val != nodes[i].old_val.b_val);
                }
                else if (nodes[i].type == 2)
                {
                    changed = (nodes[i].new_val.i_val != nodes[i].old_val.i_val);
                }
                else
                {
                    changed = (nodes[i].new_val.f_val != nodes[i].old_val.f_val);
                }

                if (changed)
                {
                    cJSON *item = cJSON_CreateObject();
                    cJSON_AddNumberToObject(item, "key", nodes[i].key);

                    if (nodes[i].type == 1)
                    {
                        cJSON_AddBoolToObject(item, "value", nodes[i].new_val.b_val);
                    }
                    else if (nodes[i].type == 2)
                    {
                        cJSON_AddNumberToObject(item, "value", nodes[i].new_val.i_val);
                    }
                    else
                    {
                        cJSON_AddNumberToObject(item, "value", nodes[i].new_val.f_val);
                    }

                    cJSON_AddItemToArray(report, item);

                    // 更新旧值
                    nodes[i].old_val = nodes[i].new_val;
                }
            }
        }

        // 如果有数据需要上报
        if (cJSON_GetArraySize(report) > 0)
        {
            send_mqtt_message(cfg, report);
        }

        cJSON_Delete(report);
        pthread_mutex_unlock(&cfg->lock);

        // 等待下次上报
        sleep(cfg->report_interval);
    }

    return NULL;
}

// 数据存储线程
void *storage_thread(void *arg)
{
    struct GlobalConfig *cfg = (struct GlobalConfig *)arg;
    sqlite3 *db;

    // 打开数据库
    if (sqlite3_open(DB_PATH, &db) != SQLITE_OK)
    {
        fprintf(stderr, "无法打开数据库: %s\n", sqlite3_errmsg(db));
        return NULL;
    }

    while (1)
    {
        sleep(30); // 每30秒存储一次数据

        pthread_mutex_lock(&cfg->lock);

        // 存储所有数据点到数据库
        for (int i = 0; i < cfg->node_count; i++)
        {
            store_data_point(db, &cfg->shm_nodes[i]);
        }

        pthread_mutex_unlock(&cfg->lock);
    }

    sqlite3_close(db);
    return NULL;
}

// 指令处理线程
void *command_thread(void *arg)
{
    struct GlobalConfig *cfg = (struct GlobalConfig *)arg;
    int topic_len;
    char *topic_name = NULL;

    while (1)
    {
        // 接收MQTT消息
        MQTTClient_message *msg = NULL;
        if (MQTTClient_receive(cfg->mqtt_client, &topic_name, &topic_len,
                               &msg, TIMEOUT) == MQTTCLIENT_SUCCESS)
        {
            if (msg)
            {
                // 确保字符串以null结尾
                char *payload = malloc(msg->payloadlen + 1);
                memcpy(payload, msg->payload, msg->payloadlen);
                payload[msg->payloadlen] = '\0';

                // 处理指令
                handle_command(cfg, payload);

                free(payload);
                MQTTClient_freeMessage(&msg);
                MQTTClient_free(topic_name);
            }
        }
    }
    return NULL;
}

int parse_config(struct GlobalConfig *config, struct std_node **nodes, int *node_count)
{
    FILE *fp = fopen(CONFIG_PATH, "r");
    if (!fp)
    {
        perror("Failed to open config file");
        return -1;
    }

    char buffer[4096]; // 增大缓冲区
    size_t len = fread(buffer, 1, sizeof(buffer) - 1, fp);
    buffer[len] = '\0';
    fclose(fp);

    cJSON *root = cJSON_Parse(buffer);
    if (!root)
    {
        fprintf(stderr, "JSON parse error: %s\n", cJSON_GetErrorPtr());
        return -1;
    }

    // 解析协议版本
    cJSON *version = cJSON_GetObjectItem(root, "version");
    if (version && version->valuestring)
    {
        if (strcmp(version->valuestring, "v1.0") == 0)
        {
            config->version = 1;
        }
    }

    // 解析上报配置
    cJSON *report = cJSON_GetObjectItem(root, "report");
    if (report)
    {
        cJSON *type = cJSON_GetObjectItem(report, "type");
        if (type)
        {
            config->report_type = (ReportMode)type->valueint;
        }
        cJSON *period = cJSON_GetObjectItem(report, "period");
        if (period)
        {
            config->report_interval = period->valueint;
        }
    }

    // 解析MQTT服务器配置
    cJSON *mqtt_server = cJSON_GetObjectItem(root, "mqtt_server");
    if (mqtt_server)
    {
        cJSON *addr = cJSON_GetObjectItem(mqtt_server, "addr");
        if (addr)
        {
            strncpy(config->mqtt_addr, addr->valuestring, sizeof(config->mqtt_addr) - 1);
        }
        cJSON *port = cJSON_GetObjectItem(mqtt_server, "port");
        if (port)
        {
            config->mqtt_port = port->valueint;
        }
    }

    // 解析modbus设备配置
    cJSON *mb_dev = cJSON_GetObjectItem(root, "mb_dev");
    if (mb_dev)
    {
        cJSON *addr = cJSON_GetObjectItem(mb_dev, "addr");
        if (addr)
        {
            strncpy(config->mb_dev_addr, addr->valuestring, sizeof(config->mb_dev_addr) - 1);
        }
        cJSON *port = cJSON_GetObjectItem(mb_dev, "port");
        if (port)
        {
            config->mb_dev_port = port->valueint;
        }
    }

    // 第一遍：统计数据点总数
    int count = 0;
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");
    if (stm32)
    {
        cJSON *data = cJSON_GetObjectItem(stm32, "data");
        if (data)
        {
            count += cJSON_GetArraySize(data);
        }
    }

    cJSON *modbus = cJSON_GetObjectItem(root, "modbus");
    if (modbus)
    {
        cJSON *data = cJSON_GetObjectItem(modbus, "data");
        if (data)
        {
            count += cJSON_GetArraySize(data);
        }
    }

    // 分配节点数组
    *nodes = (struct std_node *)malloc(count * sizeof(struct std_node));
    if (*nodes == NULL)
    {
        cJSON_Delete(root);
        return -1;
    }
    memset(*nodes, 0, count * sizeof(struct std_node));

    // 第二遍：填充节点数组
    int index = 0;
    if (stm32)
    {
        cJSON *data = cJSON_GetObjectItem(stm32, "data");
        if (data)
        {
            int size = cJSON_GetArraySize(data);
            for (int i = 0; i < size; i++)
            {
                cJSON *item = cJSON_GetArrayItem(data, i);
                struct std_node *node = &(*nodes)[index++];
                node->key = cJSON_GetObjectItem(item, "key")->valueint;
                node->type = cJSON_GetObjectItem(item, "type")->valueint;
                node->dev_type = DEV_STM32;
                node->ret = -1; // 初始状态为未采集
            }
        }
    }

    if (modbus)
    {
        cJSON *data = cJSON_GetObjectItem(modbus, "data");
        if (data)
        {
            int size = cJSON_GetArraySize(data);
            for (int i = 0; i < size; i++)
            {
                cJSON *item = cJSON_GetArrayItem(data, i);
                struct std_node *node = &(*nodes)[index++];
                node->key = cJSON_GetObjectItem(item, "key")->valueint;
                node->type = cJSON_GetObjectItem(item, "type")->valueint;
                node->dev_type = DEV_MODBUS;
                node->ret = -1; // 初始状态为未采集
            }
        }
    }

    *node_count = count;
    cJSON_Delete(root);
    return 0;
}

// 发送MQTT消息
void send_mqtt_message(struct GlobalConfig *config, cJSON *data)
{
    // 创建包含设备信息的完整消息
    // cJSON *root = cJSON_CreateObject();
    // cJSON_AddNumberToObject(root, "version", config->version);
    // cJSON_AddNumberToObject(root, "timestamp", (double)time(NULL));
    // cJSON_AddItemToObject(root, "data", data);

    // char *payload = cJSON_PrintUnformatted(root);

    cJSON *root = cJSON_CreateObject();
    cJSON_AddNumberToObject(root, "version", config->version);
    cJSON_AddNumberToObject(root, "timestamp", (double)time(NULL));

    // 使用cJSON_Duplicate复制data，而非直接添加
    cJSON_AddItemToObject(root, "data", cJSON_Duplicate(data, 1)); // 递归复制

    char *payload = cJSON_PrintUnformatted(root);

    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    pubmsg.payload = payload;
    pubmsg.payloadlen = strlen(payload);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;

    MQTTClient_deliveryToken token;
    MQTTClient_publishMessage(config->mqtt_client, TOPIC_UP, &pubmsg, &token);
    MQTTClient_waitForCompletion(config->mqtt_client, token, TIMEOUT);

    free(payload);
    cJSON_Delete(root);
}

// 处理指令
void handle_command(struct GlobalConfig *config, const char *command)
{
    cJSON *root = cJSON_Parse(command);
    if (!root)
        return;

    cJSON *cmd_type = cJSON_GetObjectItem(root, "command");
    if (!cmd_type)
    {
        cJSON_Delete(root);
        return;
    }

    const char *cmd_str = cmd_type->valuestring;
    CommandType type;

    if (strcmp(cmd_str, "control") == 0)
    {
        type = CMD_CONTROL;
    }
    else if (strcmp(cmd_str, "refresh") == 0)
    {
        type = CMD_REFRESH;
    }
    else if (strcmp(cmd_str, "change_mode") == 0)
    {
        type = CMD_CHANGE_MODE;
    }
    else if (strcmp(cmd_str, "history") == 0)
    {
        type = CMD_HISTORY_DATA;
    }
    else
    {
        cJSON_Delete(root);
        return;
    }

    pthread_mutex_lock(&config->lock);

    switch (type)
    {
    case CMD_CONTROL:
    {
        // 处理控制命令
        cJSON *key = cJSON_GetObjectItem(root, "key");
        cJSON *value = cJSON_GetObjectItem(root, "value");

        if (key && value)
        {
            for (int i = 0; i < config->node_count; i++)
            {
                if (config->shm_nodes[i].key == key->valueint)
                {
                    // 通过消息队列发送控制命令
                    char msg[256];
                    snprintf(msg, sizeof(msg), "CTRL:%d:%d",
                             key->valueint, value->valueint);

                    // 根据设备类型设置不同的消息类型
                    long msg_type = config->shm_nodes[i].dev_type;

                    // 发送消息，使用设备类型作为消息类型
                    if (msg_type == 1)
                    {
                        if (msg_queue_send("32control_queue", msg, strlen(msg) + 1, msg_type) != 0)
                        {
                            perror("消息队列发送失败");
                        }
                        else
                        {
                            printf("发送控制命令: %s (消息类型: %d)\n", msg, 1);
                        }
                    }

                    if (msg_type == 2)
                    {
                        if (msg_queue_send("salvecontrol_queue", msg, strlen(msg) + 1, msg_type) != 0)
                        {
                            perror("消息队列发送失败");
                        }
                        else
                        {
                            printf("发送控制命令: %s (消息类型: %d)\n", msg, 1);
                        }
                    }

                    break;
                }
            }
        }
        break;
    }

    case CMD_REFRESH:
    {
        // 处理刷新上报
        cJSON *report = cJSON_CreateArray();
        for (int i = 0; i < config->node_count; i++)
        {
            cJSON *item = cJSON_CreateObject();
            cJSON_AddNumberToObject(item, "key", config->shm_nodes[i].key);

            if (config->shm_nodes[i].type == 1)
            {
                cJSON_AddBoolToObject(item, "value", config->shm_nodes[i].new_val.b_val);
            }
            else if (config->shm_nodes[i].type == 2)
            {
                cJSON_AddNumberToObject(item, "value", config->shm_nodes[i].new_val.i_val);
            }
            else
            {
                cJSON_AddNumberToObject(item, "value", config->shm_nodes[i].new_val.f_val);
            }

            cJSON_AddItemToArray(report, item);
        }

        send_mqtt_message(config, report);
        cJSON_Delete(report);
        break;
    }

    case CMD_CHANGE_MODE:
    {
        // 处理模式修改
        cJSON *mode = cJSON_GetObjectItem(root, "mode");
        if (mode)
        {
            // 更新内存中的模式
            if (strcmp(mode->valuestring, "none") == 0)
            {
                config->report_type = REPORT_NONE;
            }
            else if (strcmp(mode->valuestring, "change") == 0)
            {
                config->report_type = REPORT_CHANGE;
            }
            else if (strcmp(mode->valuestring, "timed") == 0)
            {
                config->report_type = REPORT_TIMED;
            }

            // TODO: 更新配置文件
            printf("上报模式更改为: %s\n", mode->valuestring);
        }
        break;
    }

    case CMD_HISTORY_DATA:
    {
        // 处理历史数据请求
        cJSON *key_item = cJSON_GetObjectItem(root, "key");
        cJSON *start_item = cJSON_GetObjectItem(root, "start");
        cJSON *end_item = cJSON_GetObjectItem(root, "end");

        if (key_item && start_item && end_item)
        {
            cJSON *history = get_history_data(
                (time_t)start_item->valuedouble,
                (time_t)end_item->valuedouble,
                key_item->valueint);

            if (history)
            {
                send_mqtt_message(config, history);
                cJSON_Delete(history);
            }
        }
        break;
    }
    }

    pthread_mutex_unlock(&config->lock);
    cJSON_Delete(root);
}

// 初始化数据库
int init_database()
{
    sqlite3 *db;
    if (sqlite3_open(DB_PATH, &db) != SQLITE_OK)
    {
        fprintf(stderr, "Cannot open database: %s\n", sqlite3_errmsg(db));
        return -1;
    }

    char *sql = "CREATE TABLE IF NOT EXISTS history ("
                "id INTEGER PRIMARY KEY AUTOINCREMENT,"
                "timestamp INTEGER NOT NULL,"
                "key INTEGER NOT NULL,"
                "type INTEGER NOT NULL,"
                "int_value INTEGER,"
                "float_value REAL);";

    char *err_msg = 0;
    if (sqlite3_exec(db, sql, 0, 0, &err_msg) != SQLITE_OK)
    {
        fprintf(stderr, "SQL error: %s\n", err_msg);
        sqlite3_free(err_msg);
        sqlite3_close(db);
        return -1;
    }

    sqlite3_close(db);
    return 0;
}

// 存储数据点到数据库
int store_data_point(sqlite3 *db, struct std_node *node)
{
    char sql[256];
    if (node->type == 1 || node->type == 2)
    {
        snprintf(sql, sizeof(sql),
                 "INSERT INTO history (timestamp, key, type, int_value) "
                 "VALUES (%ld, %d, %d, %d);",
                 time(NULL), node->key, node->type,
                 (node->type == 1) ? node->new_val.b_val : node->new_val.i_val);
    }
    else
    {
        snprintf(sql, sizeof(sql),
                 "INSERT INTO history (timestamp, key, type, float_value) "
                 "VALUES (%ld, %d, %d, %f);",
                 time(NULL), node->key, node->type, node->new_val.f_val);
    }

    char *err_msg = 0;
    if (sqlite3_exec(db, sql, 0, 0, &err_msg) != SQLITE_OK)
    {
        fprintf(stderr, "SQL error: %s\n", err_msg);
        sqlite3_free(err_msg);
        return -1;
    }

    return 0;
}

// 获取历史数据
cJSON *get_history_data(time_t start, time_t end, int key)
{
    sqlite3 *db;
    if (sqlite3_open(DB_PATH, &db) != SQLITE_OK)
    {
        return NULL;
    }

    char sql[256];
    snprintf(sql, sizeof(sql),
             "SELECT timestamp, type, int_value, float_value FROM history "
             "WHERE key = %d AND timestamp BETWEEN %ld AND %ld "
             "ORDER BY timestamp;",
             key, start, end);

    sqlite3_stmt *stmt;
    if (sqlite3_prepare_v2(db, sql, -1, &stmt, 0) != SQLITE_OK)
    {
        sqlite3_close(db);
        return NULL;
    }

    cJSON *result = cJSON_CreateArray();

    while (sqlite3_step(stmt) == SQLITE_ROW)
    {
        cJSON *item = cJSON_CreateObject();
        time_t ts = sqlite3_column_int64(stmt, 0);
        int type = sqlite3_column_int(stmt, 1);

        cJSON_AddNumberToObject(item, "timestamp", ts);
        cJSON_AddNumberToObject(item, "key", key);

        if (type == 1 || type == 2)
        {
            int value = sqlite3_column_int(stmt, 2);
            cJSON_AddNumberToObject(item, "value", value);
        }
        else
        {
            float value = sqlite3_column_double(stmt, 3);
            cJSON_AddNumberToObject(item, "value", value);
        }

        cJSON_AddItemToArray(result, item);
    }

    sqlite3_finalize(stmt);
    sqlite3_close(db);
    return result;
}

int main(int argc, char const *argv[])
{
    // 为全局配置分配内存
    config = (struct GlobalConfig *)malloc(sizeof(struct GlobalConfig));
    if (config == NULL)
    {
        perror("malloc for GlobalConfig failed");
        return -1;
    }

    // 初始化config结构体
    memset(config, 0, sizeof(struct GlobalConfig));
    pthread_mutex_init(&config->lock, NULL); // 初始化锁

    // 解析json
    struct std_node *nodes = NULL;
    int node_count = 0;
    if (parse_config(config, &nodes, &node_count) != 0)
    {
        fprintf(stderr, "Parse config failed\n");
        free(config);
        return -1;
    }

    config->node_count = node_count;

    // 初始化共享内存
    // size_t shm_size = sizeof(struct std_node) * node_count;
    shm_init(&shm_para, "test", 2048);
    void *addr = shm_getaddr(&shm_para);
    if (addr == NULL)
    {
        fprintf(stderr, "Failed to get shared memory address\n");
        free(nodes);
        free(config);
        return -1;
    }

    // 将数据节点复制到共享内存
    config->shm_nodes = (struct std_node *)addr;
    memcpy(config->shm_nodes, nodes, sizeof(struct std_node) * node_count);
    free(nodes);

    // 创建MQTT地址
    char mqtt_address[128];
    // snprintf(mqtt_address, sizeof(mqtt_address), "tcp://%s:%d",
    //          config->mqtt_addr, config->mqtt_port);
    snprintf(mqtt_address, sizeof(mqtt_address), "tcp://localhost:%d",
             config->mqtt_port);

    // 初始化MQTT
    MQTTClient_create(&config->mqtt_client, mqtt_address, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    if (MQTTClient_connect(config->mqtt_client, &conn_opts) != MQTTCLIENT_SUCCESS)
    {
        fprintf(stderr, "Failed to connect to MQTT broker\n");
        shm_del(&shm_para);
        free(config);
        return -1;
    }

    // 订阅下行主题
    MQTTClient_subscribe(config->mqtt_client, TOPIC_DOWN, QOS);

    // 初始化数据库
    if (init_database() != 0)
    {
        fprintf(stderr, "Failed to init database\n");
        shm_del(&shm_para);
        free(config);
        return -1;
    }

    // 创建线程
    pthread_t report_tid, storage_tid, command_tid;
    // 上报线程
    if (pthread_create(&report_tid, NULL, reporting_thread, config) != 0)
    {
        perror("上报线程创建失败");
        return -1;
    }
    // 存储线程
    if (pthread_create(&storage_tid, NULL, storage_thread, config) != 0)
    {
        perror("存储线程创建失败");
        return -1;
    }
    // 指令处理线程
    if (pthread_create(&command_tid, NULL, command_thread, config) != 0)
    {
        perror("指令处理线程创建失败");
        return -1;
    }

    printf("网关已启动，共有 %d 个数据点\n", node_count);
    printf("上报模式: %d, 上报间隔: %d 秒\n", config->report_type, config->report_interval);

    // 主线程等待
    pthread_join(report_tid, NULL);
    pthread_join(storage_tid, NULL);
    pthread_join(command_tid, NULL);

    // 清理资源
    MQTTClient_disconnect(config->mqtt_client, TIMEOUT);
    MQTTClient_destroy(&config->mqtt_client);
    shm_del(&shm_para);
    pthread_mutex_destroy(&config->lock);
    free(config);

    return 0;
}